package com.chatsimple.netty.task

import com.chatsimple.logDebug
import com.chatsimple.netty.task.HeartbeatHandler.Companion.MAX_TIMESTAMP
import com.google.protobuf.ByteString
import com.ly.data.MessageProtobuf
import com.ly.data.MessageProtobuf.TaskProto
import io.netty.channel.Channel
import io.netty.channel.ChannelId
import kotlinx.coroutines.*
import java.util.concurrent.ConcurrentHashMap
import kotlin.math.abs

/**
 * 心跳处理器
 * 超[MAX_TIMESTAMP]ms未收到心跳，将主动断开连接
 */
class HeartbeatHandler : ITaskHandler {
    private val mHeartbeatJobs = ConcurrentHashMap<ChannelId, Job>()
    private val mScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    private val mJobCreateTimestamps = ConcurrentHashMap<ChannelId, Long>()

    override fun handle(channel: Channel, task: TaskProto) {
        if (task.type != MessageProtobuf.TaskType.Heartbeat) return
        val now = System.currentTimeMillis()
        val timestamp = mJobCreateTimestamps[channel.id()] ?: now
        val diff = abs(now - timestamp)
        logDebug("客户端(${channel.remoteAddress()}) 心跳时差:${diff}")
        if (diff < MAX_TIMESTAMP) {
            //未超过最大等待时间,取消等待协程，并且客户端回复
            mHeartbeatJobs.remove(channel.id())?.cancel()
            mJobCreateTimestamps.remove(channel.id())
            //重新设置时间戳后，需要给客户端回复
            val heartbeatResponse = TaskProto.newBuilder()
                .setType(MessageProtobuf.TaskType.Heartbeat)
                .setSeq(task.seq)
                .setFrom(channel.localAddress().toString())
                .setData(ByteString.EMPTY)
                .setVersion(task.version)
                .build()
            channel.writeAndFlush(heartbeatResponse).addListener {
                logDebug("心跳回复 ${it.isSuccess} $heartbeatResponse")
            }
        }

        val waitJob = mHeartbeatJobs[channel.id()]
        if (waitJob == null || !waitJob.isActive) {
            //如果有没有waitJob存在，那就创建一个
            logDebug("客户端(${channel.remoteAddress()}) 创建心跳超时Job")
            mJobCreateTimestamps[channel.id()] = now
            mHeartbeatJobs[channel.id()] = mScope.launch {
                delay(MAX_TIMESTAMP)
                channel.closeWithLog()
            }
        }

    }


    //客户端主动断开，移除计时
    override fun destroy(channel: Channel?) {
        try {
            mHeartbeatJobs.remove(channel?.id())?.cancel()
            mJobCreateTimestamps.remove(channel?.id())
        } catch (e: Exception) {
            logDebug("(${channel?.remoteAddress()})destroy error $e")
        }
    }


    private fun Channel.closeWithLog() {
        mJobCreateTimestamps.remove(id())
        close().addListener {
            logDebug("(${remoteAddress()})超过${MAX_TIMESTAMP}ms未收到心跳，关闭客户端连接 ${it.isSuccess}")
        }
    }

    companion object {
        private const val MAX_TIMESTAMP = 50 * 1000L //心跳最大时间戳
    }
}